package com.tid.hbasees.utils.hbase;

import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.tid.hbasees.utils.GenericClass;
import com.tid.hbasees.utils.JsonHelper;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.Callable;

public class ScanWorker implements Callable<List<String>> {
    private static final Logger LOG = Logger.getLogger(ScanWorker.class);

    private String tableName;
    private Scan s;
    private HBaseHelper hBaseHelper;

    public  ScanWorker(Scan s, String tableName, HBaseHelper hBaseHelper) {
        this.s = s;
        this.tableName = tableName;
        this.hBaseHelper = hBaseHelper;
    }

    private HTableInterface getTable() throws IOException {
        return hBaseHelper.getTable(tableName);
    }

    @Override
    public List<String> call() throws Exception {
        List<String> result = Lists.newLinkedList();
        try (HTableInterface table = getTable(); ResultScanner rs = table.getScanner(s)) {

            for (Result r : rs) {
                String rowkey = hBaseHelper.getStringValue(r, "rowkey");
                if (rowkey != null) {
                    result.add(rowkey);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            LOG.error("failed to scan table: " + tableName);
        }
        return result;
    }
}
